
package com.lee.job; 
 
import com.google.common.collect.Lists; 
import com.lee.enumeration.FinishQualifierEnum; 
import com.lee.info.CkFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.utils.*; 
import org.apache.flink.api.common.functions.FlatMapFunction; 
import org.apache.flink.util.Collector; 
 
import java.time.Duration; 
import java.time.LocalDateTime; 
import java.util.*; 
 
public class CreateC1Function implements FlatMapFunction<CkFlowInfo, CkFlowInfo> { 
 
 
    @Override
    public void flatMap(CkFlowInfo rawCkFlowInfo, Collector<CkFlowInfo> collector)  {
 
        try {
 
            Logger.info(rawCkFlowInfo.getDIndex(),"/",rawCkFlowInfo.getDCIndex(),"CreateC1Function ->  rawCkFlowInfo ", rawCkFlowInfo);
            LocalDateTime start = LocalDateTime.now();
 
            long index = rawCkFlowInfo.getDIndex();
            long cindex = rawCkFlowInfo.getDCIndex();
            List<String> bdsCkItemList = rawCkFlowInfo.getCkList();
 
            for (String item : bdsCkItemList) {
                CkFlowInfo ckFlowInfo = new CkFlowInfo();
                ckFlowInfo.setDIndex(index);
                ckFlowInfo.setDCIndex(cindex);
                ckFlowInfo.setCkList(Lists.newArrayList(item));
                ckFlowInfo.setDataFlow(bdsCkItemList);
 
                ckFlowInfo.getCkList().sort(String::compareTo);
 
                if (Objects.isNull(ckFlowInfo.getCkList()) || ckFlowInfo.getCkList().size() == 0) {
                    collector.collect(ckFlowInfo);
                    continue;
                }
 

 
                collector.collect(ckFlowInfo);
                Logger.debug(rawCkFlowInfo.getDIndex(),"/",rawCkFlowInfo.getDCIndex(),"CreateC1Function ->  collected ckFlowInfo ", ckFlowInfo.getCkList(),index);
 
 
            }
            DMapUtils.addFinishIndex(FinishQualifierEnum.CREATE_C1.getCode(),1l);
            LocalDateTime end = LocalDateTime.now();
            Duration between = Duration.between(start, end);
 
            Logger.info(rawCkFlowInfo.getDIndex(),"/",rawCkFlowInfo.getDCIndex(),between.toMillis(),"END CreateC1Function ->  rawCkFlowInfo ", rawCkFlowInfo);
 
 
 
        }catch (Exception e){
            Logger.error("CreateC1Function -> 初始化数据报错：",e);
            e.printStackTrace();
        }
    }
 
 
 
} 
